{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "87420f48",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Beam Notebooks Demo\n",
    "\n",
    "This example demonstrates how to set up an Apache Beam pipeline that reads from a\n",
    "[Google Cloud Storage](https://cloud.google.com/storage) file containing text from Shakespeare's work *King Lear*, \n",
    "tokenizes the text lines into individual words, and performs a frequency count on each of those words. \n",
    "\n",
    "We will perform the aggregation operations using the Beam Dataframes API, which allows us to use Pandas-like syntax to write your transformations. We will see how we can easily translate from using Pandas locally to using Dataframes in Apache Beam (which could then be run on Dataflow).\n",
    "\n",
    "We will then show how to use the `beam_sql` cell magic to use SQL to accomplish the same tasks as we performed using the Dataframes API. We will then show how to join multiple PCollections using `beam_sql`.\n",
    "\n",
    "For details about the Apache Beam Dataframe API, see the [Documentation](https://beam.apache.org/documentation/dsls/dataframes/overview/).\n",
    "\n",
    "For details about the Apache Beam SQL API, see the [Documentation](https://beam.apache.org/documentation/dsls/sql/overview/).\n",
    "\n",
    "## Getting set up and importing our data\n",
    "\n",
    "We first start with the necessary imports:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b9f352d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Python's regular expression library\n",
    "import re\n",
    "import typing\n",
    "\n",
    "# Beam and interactive Beam imports\n",
    "import apache_beam as beam\n",
    "from apache_beam.runners.interactive.interactive_runner import InteractiveRunner\n",
    "import apache_beam.runners.interactive.interactive_beam as ib\n",
    "\n",
    "# Dataframe API imports\n",
    "from apache_beam.dataframe.convert import to_dataframe\n",
    "from apache_beam.dataframe.convert import to_pcollection"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "feb53e3a",
   "metadata": {},
   "source": [
    "We will be using the `re` library to parse our lines of text. We will import the `InteractiveRunner` class for executing out pipeline in the notebook environment and the `interactive_beam` module for exploring the PCollections. Finally we will import two functions from the Dataframe API, `to_dataframe` and `to_pcollection`. `to_dataframe` converts your (schema-aware) PCollection into a dataframe and `to_pcollection` goes back in the other direction to a `PCollection` of type `beam.Row`.\n",
    "\n",
    "We will first create a composite PTransform `ReadWordsFromText` to read in a file pattern (`file_pattern`), use the `ReadFromText` source to read in the files, and then `FlatMap` with a lambda to parse the line into individual words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "71cc4abe",
   "metadata": {},
   "outputs": [],
   "source": [
    "class ReadWordsFromText(beam.PTransform):\n",
    "    \n",
    "    def __init__(self, file_pattern):\n",
    "        self._file_pattern = file_pattern\n",
    "    \n",
    "    def expand(self, pcoll):\n",
    "        return (pcoll.pipeline\n",
    "                | beam.io.ReadFromText(self._file_pattern)\n",
    "                | beam.FlatMap(lambda line: re.findall(r'[\\w\\']+', line.strip(), re.UNICODE)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7de319c",
   "metadata": {},
   "source": [
    "To be able to process our data in the notebook environment and explore the PCollections, we will use the interactive runner. We create this pipeline object in the same manner as usually, but passing in `InteractiveRunner()` as the runner."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8fa57ab4",
   "metadata": {},
   "outputs": [],
   "source": [
    "p = beam.Pipeline(InteractiveRunner())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8c1fdcdb",
   "metadata": {},
   "source": [
    "Now we're ready to start processing our data! We first apply our `ReadWordsFromText` transform to read in the lines of text from Google Cloud Storage and parse into individual words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b3d67e4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "words = p | 'ReadWordsFromText' >> ReadWordsFromText('gs://apache-beam-samples/shakespeare/kinglear.txt')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e62cb595",
   "metadata": {},
   "source": [
    "Now we will see some capabilities of the interactive runner. First we can use `ib.show` to view the contents of a specific `PCollection` from any point of our pipeline. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9286d157",
   "metadata": {},
   "outputs": [],
   "source": [
    "ib.show(words)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8be2093a",
   "metadata": {},
   "source": [
    "Great! We see that we have 28,001 words in our PCollection and we can view the words in our PCollection. \n",
    "\n",
    "We can also view the current DAG for our graph by using the `ib.show_graph()` method. Note that here we pass in the pipeline object rather than a PCollection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92c18ce6",
   "metadata": {},
   "outputs": [],
   "source": [
    "ib.show_graph(p)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d3c0802",
   "metadata": {},
   "source": [
    "In the above graph, the rectanglar boxes correspond to PTransforms and the circles correspond to PCollections. \n",
    "\n",
    "## Using the Dataframes API\n",
    "\n",
    "Next we will add a simple schema to our PCollection and convert the PCollection into a dataframe using the `to_dataframe` method. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "95bf1d2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "class WordRow(typing.NamedTuple):\n",
    "    word: str\n",
    "\n",
    "word_rows = words | 'ApplySchema' >> beam.Map(lambda word : WordRow(word=word)).with_output_types(WordRow)\n",
    "\n",
    "df = to_dataframe(word_rows)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3694d5d",
   "metadata": {},
   "source": [
    "We can now explore our PCollection as a Pandas-like dataframe! One of the first things many data scientists do as soon as they load data into a dataframe is explore the first few rows of data using the `head` method. Let's see what happens here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d963e261",
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77ba08e0",
   "metadata": {},
   "source": [
    "Notice that we got a very specific type of error! The `WontImplementError` is for Pandas methods that will not be implemented for Beam dataframes. These are methods that violate the Beam model for one reason or another. For example, in this case the `head` method depends on the order of the dataframe. However, this is in conflict with the Beam model. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a04cc1b",
   "metadata": {},
   "source": [
    "Our goal however is to count the number of times each word appears in the ingested text. First we will add a new column in our dataframe named `count` with a value of `1` for all rows. After that, we will group by the value of the `word` column and apply the `sum` method for the `count` field."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "467c8822",
   "metadata": {},
   "outputs": [],
   "source": [
    "df['count'] = 1\n",
    "counted = df.groupby('word').sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9ea4ccdf",
   "metadata": {},
   "source": [
    "That's it! It looks exactly like the code one would write when using Pandas. However, what does this look like in the DAG for the pipeline? We can see this by executing `ib.show_graph(p)` as before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e79fbd2c",
   "metadata": {},
   "outputs": [],
   "source": [
    "ib.show_graph(p)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fdf3c807",
   "metadata": {},
   "source": [
    "We can see that the dataframe manipulations added a new PTransform to our pipeline. Let us convert the dataframe back to a PCollection so we can use `ib.show` to view the contents."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ee7367e",
   "metadata": {},
   "outputs": [],
   "source": [
    "word_counts = to_pcollection(counted, include_indexes=True)\n",
    "ib.show(word_counts)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9afd928f",
   "metadata": {},
   "source": [
    "Great! We can now see that the words have been successfully counted. Finally let us build in a sink into the pipeline. We can do this in two ways. If we wish to write to a CSV file, then we can use the dataframe's `to_csv` method. We can also use the `WriteToText` transform after converting back to a PCollection. Let's do both and explore the outputs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "50f5f821",
   "metadata": {},
   "outputs": [],
   "source": [
    "counted.to_csv('from_df.csv')\n",
    "_ = word_counts | beam.io.WriteToText('from_pcoll.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e225c3b7-c2f4-4610-8ed0-0a2b4c4c1c32",
   "metadata": {},
   "source": [
    "Before saving the outputs to the sinks, let's take a peek at our finished pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56cb1509",
   "metadata": {},
   "outputs": [],
   "source": [
    "ib.show_graph(p)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b22b0b07",
   "metadata": {},
   "source": [
    "Note that we can see the branching with two different sinks, also we can see where the dataframe is converted back to a PCollection. We can run our entire pipeline by using `p.run()` as normal."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3e365786",
   "metadata": {},
   "outputs": [],
   "source": [
    "p.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "601272b9",
   "metadata": {},
   "source": [
    "Let us now look at the beginning of the CSV files using the bash line magic with the `head` command to compare."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb0407fb",
   "metadata": {},
   "outputs": [],
   "source": [
    "!head from_df*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f6ae2b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "!head from_pcoll*"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5cd1e97",
   "metadata": {},
   "source": [
    "We (functionally) end up with the same information as expected! The big difference is in how the results are presented. In the case of the output from the `WriteToText` connector, we did not convert our PCollection from objects of type `Row`. We could write a simple intermediate transform to pull out the properties of the `Row` object into a comma-seperated representation. For example:\n",
    "\n",
    "```\n",
    "def row_to_csv(element):\n",
    "    output = f\"{element.word},{element.count}\"\n",
    "    return output\n",
    "```\n",
    "\n",
    "The we could replace the code `_ = word_counts | beam.io.WriteToText('from_pcoll.csv')` with\n",
    "\n",
    "```\n",
    "_ = word_counts | beam.Map(row_to_csv)\n",
    "                | beam.io.WriteToText('from_pcoll.csv')\n",
    "```\n",
    "\n",
    "However, note that the `to_csv` method for the dataframe took care of this conversion for us."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d8fc0527-2283-447e-98d8-33e8b1931068",
   "metadata": {},
   "source": [
    "## Setting up to use Beam SQL in notebooks\n",
    "\n",
    "Now we will accomplish the same task, counting the number of instances of each word in _King Lear_, using Beam SQL. Recall that in Python, the `SqlTransform` PTransform is a cross-language transform written in Java. Let us be sure that we have both Docker and Java installed in our instance so that we can leverage cross-language transforms."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6522dcb9-5a74-4d4d-8c53-9d092fb83320",
   "metadata": {},
   "outputs": [],
   "source": [
    "!docker image list\n",
    "!java -version"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25b4c0e7-aa96-4d76-bc02-deb15048149e",
   "metadata": {},
   "source": [
    "If you are running this notebook using Vertex AI Workbench in a Beam Notebook instance, thenn you should see a list of Docker images that are locally available and that we are using OpenJDK v.1.8+. If you running this notebook elsewhere, then you will need to ensure that Docker and Java are installed and that the right JARs and containers are built. The Appendix of this notebook will walk you through the steps you need to follow to build the relevant JARs and containers."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce20148a-859b-4a48-b65f-5f6ad7abf085",
   "metadata": {},
   "source": [
    "[Beam SQL](https://beam.apache.org/documentation/dsls/sql/overview/) allows a Beam user to query PCollections with SQL statements. Currently, `InteractiveRunner` does not support `SqlTransform` directly. However, a user could use the `beam_sql` magic to run Beam SQL in the notebook and introspect the result. \n",
    "\n",
    "`beam_sql` is an IPython [custom magic](https://ipython.readthedocs.io/en/stable/config/custommagics.html). If you're not familiar with magics, here are some [built-in examples](https://ipython.readthedocs.io/en/stable/interactive/magics.html). It's a convenient way to validate your queries locally against known/test data sources when prototyping a Beam pipeline with SQL, before productionizing it on remote cluster/services.\n",
    "\n",
    "This Beam Notebook environment has preloaded the `beam_sql` magic. You can also explicitly load it via `%load_ext apache_beam.runners.interactive.sql.beam_sql_magics` if you set up your own notebook elsewhere.\n",
    "\n",
    "The `beam_sql` magic can be used as either a line magic or a cell magic. You can check its usage by running:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e00e084-f543-43ac-b5ce-7a38de1180b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "%beam_sql -h"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "38b52293-98c2-4dd7-af9e-ff2809a5300d",
   "metadata": {},
   "source": [
    "Why would you want to use `beam_sql` in a notebook environment? \n",
    "- You can leverage an inutitive syntax with SQL.\n",
    "    - No need to use the constant `PCOLLECTION` when querying a single PCollection\n",
    "    - No need to name multiple input PCollections, instead you can refer to them by their variable names.\n",
    "- No need to write `SqlTransform` and other Beam related boilerplate code.\n",
    "- You can introspect the result immediately.\n",
    "- Coder registration for your PCollection schemas is handled automatically."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "16c2d1e7-8763-4365-84b8-3b8e868d3bc4",
   "metadata": {},
   "source": [
    "## Using `beam_sql` \n",
    "\n",
    "First let us revisit our earlier problem. We want to count the number of times each word appears in King Lear. Instead of using the Dataframe API, let's use SQL to accomplish the same task as before. Note that as of the time of writing, there is no native Calcite SQL function to break a string into words as we have using the `re` library. For that reason, we will start with the PCollection we called `word_rows` before. Note that we need a subclass of `NamedTuple` to use the `beam_sql` cell magic, which is why we're starting here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f86b07ae-0c68-4514-9a8c-2f7e3454b207",
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "#Set logging level to ERROR to minimize logs in notebook.\n",
    "logging.root.setLevel(logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e4906ca2-3063-4eff-aa27-b3c08327a305",
   "metadata": {},
   "source": [
    "Let's start out by looking at ten of the elements in our `word_rows` PCollection.\n",
    "\n",
    "**Note:** The first time you execute the `beam_sql` cell magic in a notebook, it will take a few minutes to run. The relevant container for executing the cross-language transform needs to first be built."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b58e67d-4de8-4667-8e17-13f3e3e44f07",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%beam_sql -o ten_words\n",
    "SELECT * FROM word_rows LIMIT 10"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7835b73-104d-43f4-8ab0-a6085256fe7c",
   "metadata": {},
   "source": [
    "Great! Everything is working as we can see ten words from our PCollection. Now we can quickly perform the same aggregation as before, but now using SQL."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f8352e64-04bb-40f3-83f9-a395b0bb08c3",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%beam_sql -o word_count\n",
    "SELECT word, COUNT(*) AS word_count FROM word_rows GROUP BY word "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2dd83f6f-5686-4991-adbb-1c314d8f7867",
   "metadata": {},
   "source": [
    "We saved the output of the SQL query in a new PCollection called `word_count`. We can view this using the Interactive Runner as before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "515c55c4-4730-4887-b3b2-cbaae545f1b8",
   "metadata": {},
   "outputs": [],
   "source": [
    "ib.show(word_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f8987364-692e-47b4-b6c7-44c6686a5415",
   "metadata": {},
   "source": [
    "## Joins using `beam_sql`\n",
    "\n",
    "Now let us look at one more example of using SQL in the notebook environment. In particular, let us see how easy it is to join two PCollections! We will create two PCollection in-memory for this example: One consisting of PCollections with schema of type `Person` and one consisting of type `Pet`. For the sake of simplicity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8e3f36e-0139-4361-9925-79176c5b0b2c",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Person(typing.NamedTuple):\n",
    "    name: str\n",
    "    age: int\n",
    "\n",
    "class Pet(typing.NamedTuple):\n",
    "    name: str\n",
    "    owner_name: str\n",
    "    species: str"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7a1af7ff-8cc5-4f4c-812e-3c72213e60c3",
   "metadata": {},
   "source": [
    "Now we will create a new pipeline and the corresponding PCollections."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2193efd-c8e6-435a-8b89-e3d0973df634",
   "metadata": {},
   "outputs": [],
   "source": [
    "p_join = beam.Pipeline(InteractiveRunner())\n",
    "\n",
    "people = p_join | \"Create_People\" >> beam.Create([Person('Bob', 19), \n",
    "                                                  Person('Alice', 42),\n",
    "                                                  Person('Ted', 26),\n",
    "                                                  Person('Michael', 29)])\n",
    "\n",
    "pets = p_join | \"Create_Pets\" >> beam.Create([Pet('Cooper', 'Michael', 'Dog'),\n",
    "                                              Pet('Moose', 'Alice', 'Gerbil'),\n",
    "                                              Pet('The Destroyer', 'Alice', 'Cat'),\n",
    "                                              Pet('Ted Jr.', 'Ted', 'Turtle'),\n",
    "                                              Pet('Felix', 'Ted', 'Cat')])\n",
    "                        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4d037998-9851-4cb7-a13b-fea9295aeaac",
   "metadata": {},
   "source": [
    "Now we have created our sample PCollections. Suppose that we want to answer the following question: What is the average owner age for each species of pet? We have (reasonably) three options we could use:\n",
    "1. Create KV pairs with owner names as the key, perform a `CoGroupByKey`, extract the `species` and `age` fields, and the perform `beam.CombinePerKey()` with an average `CombineFn` using `species` as the key.\n",
    "2. Use the Dataframe API to join the dataframes corresponding to each PCollection, join the dataframes on the `age` and `owner_age` columns, drop the uncessary fields and use the `.groupby()` method followed by the `.count()` method to aggregate.\n",
    "3. Write a SQL query!\n",
    "\n",
    "The dataframe option is definitely easier than using `CoGroupByKey`...but SQL is even easier than the other two options! Let's now use the `beam_sql` magic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "41583942-147e-42fc-a5f7-a1bbc3687ed9",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%beam_sql -o age_per_species\n",
    "SELECT \n",
    "    pets.species as species, \n",
    "    AVG(people.age) as average_age\n",
    "FROM \n",
    "    people \n",
    "JOIN \n",
    "    pets \n",
    "ON \n",
    "    people.name = pets.owner_name\n",
    "GROUP BY \n",
    "    pets.species"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18c4e433-7369-42fb-a884-7d35647a683b",
   "metadata": {},
   "source": [
    "What if we want to execute this pipeline on Dataflow? We can use the option `-r DataflowRunner` when using the `beam_sql` magic. When you execute the query with this option set, a pop-up will appear for a minimal set of options to run the pipeline and you will be informed where the results of the query will be stored."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eba92189-63dc-45a4-957f-0e8818de1aaa",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%beam_sql -o age_per_species -r DataflowRunner\n",
    "SELECT \n",
    "    pets.species as species, \n",
    "    AVG(people.age) as average_age\n",
    "FROM \n",
    "    people \n",
    "JOIN \n",
    "    pets \n",
    "ON \n",
    "    people.name = pets.owner_name\n",
    "GROUP BY \n",
    "    pets.species"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "946d4ed3-95ca-496e-93cc-136c4b8bdd15",
   "metadata": {},
   "source": [
    "Finally, we can check the output of the pipeline once it is completed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0c24599a-351f-4715-8d89-12151c967a74",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "BUCKET = 'your-bucket-name-here' #REPLACE WITH YOUR BUCKET NAME\n",
    "os.environ['BUCKET'] = BUCKET\n",
    "\n",
    "!gcloud storage cat gs://$BUCKET/staging/age_per_species*"   ]
  },
  {
   "cell_type": "markdown",
   "id": "abbb2e90-956e-4898-a70b-ccd221bf41aa",
   "metadata": {},
   "source": [
    "## Appendix: Setting up `beam_sql` locally\n",
    "\n",
    "**Important**: If you're using Beam built from your local source code, additionally:\n",
    "\n",
    "- Have the Java expansion service shadowjar built. Go to the root directory of your local beam repo and then execute:\n",
    "  `./gradlew :sdks:java:extensions:sql:expansion-service:shadowJar`.\n",
    "- Based on your jdk version, pull the docker image `docker pull apache/beam_java11_sdk` or `docker pull apache/beam_java8_sdk`.\n",
    "- Then tag the image with your current Beam dev version.  You can check the dev version under `apache_beam.version.__version__`. For example, if you're using jdk11 and dev version is `x.x.x.dev`, execute `docker image tag apache/beam_java11_sdk:latest apache/beam_java11_sdk:x.x.x.dev`."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "01. Apache Beam 2.40.0 for Python 3",
   "language": "python",
   "name": "01-apache-beam-2.40.0"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
